package com.chengzhi.nettycommoms.disruptor;

import com.chengzhi.nettycommoms.dto.NettyMessageWapper;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

public class RingBufferFactory {
    private static class SingletonHolder {
        static final RingBufferFactory instance = new RingBufferFactory();
    }

    private RingBufferFactory() {
    }

    public static RingBufferFactory instance() {
        return SingletonHolder.instance;
    }

    private static Map<String, MessageProducer> messageProducerMap = new ConcurrentHashMap<>();

    private static Map<String, MessageConsumer> messageConsumerMap = new ConcurrentHashMap<>();

    private RingBuffer<NettyMessageWapper> ringBuffer;

    private SequenceBarrier sequenceBarrier;

    private WorkerPool<NettyMessageWapper> workerPool;

    public void init(ProducerType type, int bufferSize, WaitStrategy waitStrategy, MessageConsumer[] messageConsumers) {
        this.ringBuffer = RingBuffer.create(type, new EventFactory<NettyMessageWapper>() {
            @Override
            public NettyMessageWapper newInstance() {
                return new NettyMessageWapper();
            }
        }, bufferSize, waitStrategy);
        this.sequenceBarrier = this.ringBuffer.newBarrier();
        this.workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, new ExceptionHandler<NettyMessageWapper>() {
            @Override
            public void handleEventException(Throwable throwable, long l, NettyMessageWapper nettyMessageWapper) {

            }

            @Override
            public void handleOnStartException(Throwable throwable) {
            }

            @Override
            public void handleOnShutdownException(Throwable throwable) {
            }
        }, messageConsumers);
        for (MessageConsumer messageConsumer : messageConsumers) {
            messageConsumerMap.put(messageConsumer.getConsumerId(), messageConsumer);
        }
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
        this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));
    }

    public MessageProducer getMessageProducer(String producerId) {
        MessageProducer messageProducer = messageProducerMap.get(producerId);
        if (null == messageProducer) {
            messageProducer = new MessageProducer(producerId, this.ringBuffer);
            messageProducerMap.put(producerId, messageProducer);
        }
        return messageProducer;
    }
}
